package mq

import (
	"fmt"
	"github.com/streadway/amqp"
	"logstash/config"
	"logstash/mq/queue"
	"logstash/utile"
	"time"
)

// 队列管理者
type MqMgr struct {
	ProducerMap map[string]Producer
	ConsumerMap map[string]Consumer
}

// 定义队列交换机对象
type QueueExchange struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	QuName  string // 队列名称
	RtKey   string // key值
	ExName  string // 交换机名称
	ExType  string // 交换机类型
}

//生产者
type Producer interface {
	Send(msg string) error
	getName() string
	connect() (*amqp.Connection, error)
}

//消费者
type Consumer interface {
	Consume() error
	SetId()
	GetId() string
	getName() string
	connect() error
	close() error
}

var mqMgr = &MqMgr{}

//创建队列对象
func NewMqManager() *MqMgr {
	mqMgr.ProducerMap = make(map[string]Producer, 0)
	mqMgr.ConsumerMap = make(map[string]Consumer, 0)
	return mqMgr
}

func GetMqManager() *MqMgr {
	return mqMgr
}

func GetProducer(name string) (Producer, bool) {
	v, ok := mqMgr.ProducerMap[name]
	return v, ok
}

func GetConsumer(name string) (Consumer, bool) {
	v, ok := mqMgr.ConsumerMap[name]
	return v, ok
}

func (m *MqMgr) RegisterProducer(p Producer) {
	name := p.getName()
	m.ProducerMap[name] = p
}

func (m *MqMgr) RegisterConsumer(c Consumer) {
	name := c.getName()
	m.ConsumerMap[name] = c
}

//批量注册生产者
func (m *MqMgr) RegisterProducers(mapProducers map[string]Producer) {
	for k, v := range mapProducers {
		m.ProducerMap[k] = v
	}
}

//批量注册消费者
func (m *MqMgr) RegisterConsumers(mapConsumers map[string]Consumer) {
	for k, v := range mapConsumers {
		m.ConsumerMap[k] = v
	}
}

func (m *MqMgr) HeartBeat() {
	flag := true
	conf := config.GetConfig()
	for {
		time.Sleep(time.Second * 5)
		producer, ok := m.ProducerMap[queue.PRODUCER_HEARTBEAT]
		if !ok {
			panic(fmt.Sprintf("[ "+conf.Data.Mq.Host+" ]没有注册MQ心跳检测队列：%s", queue.PRODUCER_HEARTBEAT))
		}
		if err := producer.Send("ok"); err != nil {
			if flag {
				// 发送邮件给管理员，提醒MQ挂了
				utile.SendMail2Admin("MQ [ "+conf.Data.Mq.Host+" ] Heart Beat Failed", err.Error())
			}
			flag = false
		} else {
			fmt.Println("MQ [ " + conf.Data.Mq.Host + " ] Heart Beat SUCC")
			if !flag {
				// 发送邮件给管理员，提醒MQ启动了
				utile.SendMail2Admin("MQ [ "+conf.Data.Mq.Host+" ] Heart Beat SUCC", "MQ重新检测SUCC")
			}
			flag = true
		}
	}
}

func (m *MqMgr) Start() {
	if err := recover(); err != nil {
		time.Sleep(3 * time.Second)
		fmt.Println("MQ Start 休息3秒")
		m.Start()
	}
	for _, v := range m.ConsumerMap {
		v.SetId()
		go v.Consume()
	}
}
